package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo24PageRank {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setMaster("local")

    conf.setAppName("Demo24PageRank")

    val sc = new SparkContext(conf)

    //读取数据
    val dataRDD: RDD[String] = sc.textFile("data/pagerank.txt")

    //整理i数据
    val pageLinkRDD: RDD[(String, List[String])] = dataRDD.map(line => {
      val split: Array[String] = line.split("-")
      //当前网页
      val page: String = split(0)
      //出链列表
      val link: List[String] = split(1).split(",").toList
      (page, link)
    })

    //给每一个网页一个初始的PR值
    var pageRankRDD: RDD[(String, List[String], Double)] = pageLinkRDD.map {
      case (page: String, link: List[String]) =>
        (page, link, 1.0)
    }

    var flag = true

    val Q = 0.85
    val N: Long = pageRankRDD.count()

    /**
     * 循环计算pr,
     * 下一次计算时，使用上一次结算的结果进行计算
     *
     */
    while (flag) {
      //将网页的pr值平分给出链列表
      val pinPrRDD: RDD[(String, Double)] = pageRankRDD.flatMap {
        case (_: String, link: List[String], pr: Double) =>
          //分给每一个出链列表的pr值
          val pinPr: Double = pr / link.length

          link.map(pg => (pg, pinPr))
      }
      //计算新的PR值
      val newPrRDD: RDD[(String, Double)] = pinPrRDD
        .reduceByKey(_ + _)
        .map {
          case (page: String, pr: Double) =>
            //增加阻尼系数
            val newPr: Double = (1 - Q) / N + Q * pr
            (page, newPr)
        }


      /**
       * 计算收敛条件
       * 计算当前所有网页和上一次所有网页pr值的差值平均值
       *
       */
      val valueRDD: RDD[Double] = pageRankRDD
        .map {
          case (page: String, _: List[String], pr: Double) =>
            (page, pr)
        }
        .join(newPrRDD)
        .map {
          case (_: String, (lastPr: Double, newPr: Double)) =>
            //计算差值
            Math.abs(lastPr - newPr)
        }

      //计算平均值
      val avgCha: Double = valueRDD.sum() / valueRDD.count()
      println(s"差值平均值：$avgCha")
      if (avgCha < 0.001) {
        flag = false
      }

      //关联获取出链列表
      val joinRDD: RDD[(String, (Double, List[String]))] = newPrRDD.join(pageLinkRDD)

      //将新的pr传递到下一次计算使用
      pageRankRDD = joinRDD.map {
        case (page: String, (pr: Double, link: List[String])) =>
          (page, link, pr)
      }
    }

    pageRankRDD.foreach(println)


  }

}
